package tech.treec.smartlight.service;

import android.annotation.SuppressLint;
import android.app.AlarmManager;
import android.app.Notification;
import android.app.NotificationChannel;
import android.app.NotificationManager;
import android.app.PendingIntent;
import android.app.Service;
import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.content.SharedPreferences;
import android.net.ConnectivityManager;
import android.os.Binder;
import android.os.Build;
import android.os.Handler;
import android.os.IBinder;
import android.os.Looper;
import android.os.Message;
import android.os.PowerManager;
import android.os.PowerManager.WakeLock;
import android.preference.PreferenceManager;
import android.provider.Settings;
import android.provider.Settings.Secure;
import android.util.Log;

import androidx.core.app.NotificationCompat;

import com.alibaba.fastjson.JSONObject;
import com.ibm.mqtt.IMqttClient;
import com.ibm.mqtt.MqttClient;
import com.ibm.mqtt.MqttException;
import com.ibm.mqtt.MqttNotConnectedException;
import com.ibm.mqtt.MqttPersistence;
import com.ibm.mqtt.MqttPersistenceException;
import com.ibm.mqtt.MqttSimpleCallback;
import com.young.mylib.utils.ConverterUtils;

import java.lang.ref.WeakReference;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.TimeZone;
import java.util.Timer;
import java.util.TimerTask;

import tech.treec.smartlight.Constants;
import tech.treec.smartlight.R;

public class MQTT1Service extends Service implements MqttSimpleCallback {
    private final static String TAG = "mqtt.MQTTService";
    public static final String MQTT_STATUS_INTENT = "org.mosquitto.android.mqtt.STATUS";
    public static final String MQTT_STATUS_MSG = "org.mosquitto.android.mqtt.STATUS_MSG";
    public static final String MQTT_PING_ACTION = "org.mosquitto.android.mqtt.PING";
    public static final String ACTION_MQTTSERVICE_RECEIVE = "org.mosquitto.android.mqtt.ACTION_MQTTSERVICE_RECEIVE";
    public static final String ACTION_MQTT_MESSAGE_ARRIVED = "tech.treec.smartlight_ACTION_MQTT_MESSAGE_ARRIVED";


    // constants used to define MQTT connection status
    public enum MQTTConnectionStatus {
        INITIAL,                            // initial status
        CONNECTING,                         // attempting to connect
        CONNECTED,                          // connected
        NOTCONNECTED_WAITINGFORINTERNET,    // can't connect because the phone
        //	 does not have Internet access
        NOTCONNECTED_USERDISCONNECT,        // user has explicitly requested
        //	 disconnection
        NOTCONNECTED_DATADISABLED,          // can't connect because the user
        //	 has disabled data access
        NOTCONNECTED_UNKNOWNREASON          // failed to connect for some reason
    }

    public static final int MAX_MQTT_CLIENTID_LENGTH = 22;

    private MQTTConnectionStatus connectionStatus = MQTTConnectionStatus.INITIAL;

    private String brokerHostName = "";
    private int brokerPortNumber = 1883;
    private String subTopic = "";

    private MqttPersistence usePersistence = null;
    private boolean cleanStart = false;
    private int[] qualitiesOfService = {0};

    private short keepAliveSeconds = 20 * 60;
    private String mqttClientId = null;

    private IMqttClient mqttClient = null;
    private NetworkConnectionIntentReceiver netConnReceiver;
    private BackgroundDataChangeIntentReceiver dataEnabledReceiver;
    private MsgReceiver msgReceiver;
    private PingSender pingSender;
    private Timer serviceTimer = null;
    protected SharedPreferences settings;

    public MQTT1Service() {

    }

    private final class ServiceHandler extends Handler {
        public ServiceHandler(Looper looper) {
            super(looper);
        }

        @Override
        public void handleMessage(Message msg) {
            // 处理收到的消息
        }
    }
    @Override
    public void onCreate() {
        super.onCreate();
        Log.i(TAG, "mqtt 服务建立");
        // reset status variable to initial state
        mBinder = new LocalBinder<MQTT1Service>(this);

        settings = PreferenceManager.getDefaultSharedPreferences(this);

        msgReceiver = new MsgReceiver();
        IntentFilter intentFilter = new IntentFilter();
        intentFilter.addAction(ACTION_MQTTSERVICE_RECEIVE);
        registerReceiver(msgReceiver, intentFilter);

        serviceTimer = new Timer();
        serviceTimer.schedule(timerTask, 6000, 30000);


    }

    private void initMqttService() {
        Log.i(TAG,"IS CONNECTED"+isAlreadyConnected());
        if (isAlreadyConnected()) {
            disconnect();
        }
        SharedPreferences prefs = androidx.preference.PreferenceManager.getDefaultSharedPreferences(getApplicationContext());

        brokerHostName = prefs.getString("mqttUrl", "192.168.119.60"); // 获取默认值作为备选方案。
        String port = prefs.getString("mqttPort", "1883"); // 获取默认值作为备选方案。
        brokerPortNumber = ConverterUtils.str2int(port, 1883);
        String panelId = prefs.getString("panelId", "588"); // 获取默认值作为备选方案。
        subTopic = "/light/DARMAO000" + panelId + "0000/data/report";

        connectionStatus = MQTTConnectionStatus.INITIAL;
        Log.i(TAG, "brokerHostName: " + brokerHostName + " ;brokerPortNumber: " + brokerPortNumber);
        if (dataEnabledReceiver == null) {
            dataEnabledReceiver = new BackgroundDataChangeIntentReceiver();
            registerReceiver(dataEnabledReceiver,
                    new IntentFilter(ConnectivityManager.ACTION_BACKGROUND_DATA_SETTING_CHANGED));

        }
        defineConnectionToBroker(brokerHostName);
    }

    TimerTask timerTask = new TimerTask() {
        @Override
        public void run() {
            //  Log.i(TAG, "MQTT CLIENT IS NULL?" + (mqttClient == null));
            if (mqttClient == null) {
                return;
            }
            heartbeat();
            if (!mqttClient.isConnected()) {
                defineConnectionToBroker(brokerHostName);
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        Log.i(TAG, "reHandleStart");
                        handleStart();
                    }
                }, "MQTTservice").start();
            }
            /**
             * 检查发送MAP有没有发送后未收到回复的
             */

        }
    };

    private void heartbeat() {
        //   Log.i(TAG, "heartbeat:" + isAlreadyConnected());
        if (isAlreadyConnected()) {
            SharedPreferences prefs = androidx.preference.PreferenceManager.getDefaultSharedPreferences(this);
            String panelId = prefs.getString("panelId", "588"); // 获取默认值作为备选方案。
            String topic = panelId + "/heartbeat";
            JSONObject json = new JSONObject();
            json.put("panelId", panelId);

            long ts = System.currentTimeMillis();
            Date time = new Date(ts);
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            format.setTimeZone(TimeZone.getTimeZone("GMT+8"));
            json.put("time", format.format(time));
//            json["panelId"]=panelId;
            publishMqttMessage(topic, json.toJSONString());
        }
    }

    @Override
    public void onStart(final Intent intent, final int startId) {
        Log.i(TAG, "onStart");
    }

    @Override
    public int onStartCommand(final Intent intent, int flags, final int startId) {
        Log.i(TAG, "onStartCommand");
        startForeground(1, getNotfication());
        return START_STICKY;
    }

    private Notification getNotfication() {
        String channelId = "smart_light_channel";
        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
            NotificationChannel channel = new NotificationChannel(channelId, "SMART LIGHT CHANNEL", NotificationManager.IMPORTANCE_DEFAULT);
            NotificationManager manager = getSystemService(NotificationManager.class);
            manager.createNotificationChannel(channel);
        }
        NotificationCompat.Builder builder = new NotificationCompat.Builder(this, channelId)
                .setContentTitle("Service Running")
                .setContentText("MyService is running in the foreground")
                .setSmallIcon(R.drawable.app);
        return builder.build();

    }

    synchronized void handleStart(Intent intent, int startId) {
        // before we start - check for a couple of reasons why we should stop
        handleStart();
    }

    synchronized void handleStart() {

        if (mqttClient == null) {
            Log.i(TAG,"mqtt is null");
            stopSelf();
            return;
        }


        ConnectivityManager cm = (ConnectivityManager) getSystemService(CONNECTIVITY_SERVICE);
        if (cm.getBackgroundDataSetting() == false) // respect the user's request not to use data!
        {
            connectionStatus = MQTTConnectionStatus.NOTCONNECTED_DATADISABLED;
            Log.i(TAG, "Not connected - background data disabled");
            broadcastServiceStatus("Not connected - background data disabled");
            return;
        }

        rebroadcastStatus();
        rebroadcastReceivedMessages();


        if (isOnline()) {
            if (connectToBroker()) {
                subscribeToTopic(subTopic);
            }
        } else {
            connectionStatus = MQTTConnectionStatus.NOTCONNECTED_WAITINGFORINTERNET;
            broadcastServiceStatus("Waiting for network connection");
        }
        // }

        if (netConnReceiver == null) {
            netConnReceiver = new NetworkConnectionIntentReceiver();
            registerReceiver(netConnReceiver,
                    new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));

        }
        if (pingSender == null) {
            pingSender = new PingSender();
            registerReceiver(pingSender, new IntentFilter(MQTT_PING_ACTION));
        }

        Log.i(TAG, "MQTT info: " + brokerHostName + ":" + brokerPortNumber + ";Topic: " + subTopic);
        Log.i(TAG, "isAlreadyConnected ? " + isAlreadyConnected());


    }

    @Override
    public void onDestroy() {
        super.onDestroy();
        disconnectFromBroker();
        broadcastServiceStatus("Disconnected");
        if (dataEnabledReceiver != null) {
            unregisterReceiver(dataEnabledReceiver);
            dataEnabledReceiver = null;

        }
        if (msgReceiver != null) {
            unregisterReceiver(msgReceiver);
        }

        if (mBinder != null) {
            mBinder.close();
            mBinder = null;
        }
    }

    private void broadcastServiceStatus(String statusDescription) {

        Intent broadcastIntent = new Intent();
        broadcastIntent.setAction(MQTT_STATUS_INTENT);
        broadcastIntent.putExtra(MQTT_STATUS_MSG, statusDescription);
        sendBroadcast(broadcastIntent);
    }

    private LocalBinder<MQTT1Service> mBinder;

    @Override
    public IBinder onBind(Intent intent) {
        return mBinder;
    }

    public class LocalBinder<S> extends Binder {
        private WeakReference<S> mService;

        public LocalBinder(S service) {
            mService = new WeakReference<S>(service);
        }

        public S getService() {
            return mService.get();
        }

        public void close() {
            mService = null;
        }
    }

    public MQTTConnectionStatus getConnectionStatus() {
        return connectionStatus;
    }

    public void rebroadcastStatus() {
        String status = "";

        switch (connectionStatus) {
            case INITIAL:
                status = "Please wait";
                break;
            case CONNECTING:
                status = "Connecting...";
                break;
            case CONNECTED:
                status = "Connected";
                break;
            case NOTCONNECTED_UNKNOWNREASON:
                status = "Not connected - waiting for network connection";
                break;
            case NOTCONNECTED_USERDISCONNECT:
                status = "Disconnected";
                break;
            case NOTCONNECTED_DATADISABLED:
                status = "Not connected - background data disabled";
                break;
            case NOTCONNECTED_WAITINGFORINTERNET:
                status = "Unable to connect";
                break;
        }

        //
        // inform the app that the Service has successfully connected
        broadcastServiceStatus(status);
    }

    public void disconnect() {
        disconnectFromBroker();

        // set status
        connectionStatus = MQTTConnectionStatus.NOTCONNECTED_USERDISCONNECT;

        // inform the app that the app has successfully disconnected
        broadcastServiceStatus("Disconnected");
    }

    public void connectionLost() throws Exception {
        Log.i(TAG, "connectionLost");

        if (isOnline() == false) {
            connectionStatus = MQTTConnectionStatus.NOTCONNECTED_WAITINGFORINTERNET;
            broadcastServiceStatus("Connection lost - no network connection");
        } else {

            connectionStatus = MQTTConnectionStatus.NOTCONNECTED_UNKNOWNREASON;

            broadcastServiceStatus("Connection lost - reconnecting...");
            if (connectToBroker()) {
                Log.i(TAG, "Connect lost");
                subscribeToTopic(subTopic);
            }
        }
    }

    private void defineConnectionToBroker(String brokerHostName) {
        String mqttConnSpec = "tcp://" + brokerHostName + "@" + brokerPortNumber;

        try {
            // define the connection to the broker
            mqttClient = MqttClient.createMqttClient(mqttConnSpec, usePersistence);
            // register this client app has being able to onReceiveFile messages
            mqttClient.registerSimpleHandler(this);
        } catch (MqttException e) {
            // something went wrong!
            mqttClient = null;
            connectionStatus = MQTTConnectionStatus.NOTCONNECTED_UNKNOWNREASON;
            broadcastServiceStatus("Invalid connection parameters");
        }
    }

    /*
     * (Re-)connect to the message broker
     */
    private boolean connectToBroker() {
        try {
            // try to connect
            if (mqttClient == null) {

            }
            mqttClient.connect(generateClientId(), cleanStart, keepAliveSeconds);
            broadcastServiceStatus("Connected");
            connectionStatus = MQTTConnectionStatus.CONNECTED;

            scheduleNextPing();

            return true;
        } catch (MqttException e) {
            connectionStatus = MQTTConnectionStatus.NOTCONNECTED_UNKNOWNREASON;
            broadcastServiceStatus("Unable to connect");
            scheduleNextPing();

            return false;
        }
    }

    private void subscribeToTopic(final String topicName) {
        if (isAlreadyConnected() == false) {
            Log.e("mqtt", "Unable to subscribe as we are not connected");
        } else {
            new Thread() {
                public void run() {
                    boolean subscribed = false;
                    try {
                        if (topicName.isEmpty()) {
                            Log.i(TAG, "Topic Is Null");
                            broadcastServiceStatus("Unable to subscribe");
                            return;
                        }
                        String[] topics = {topicName};
                        mqttClient.subscribe(topics, qualitiesOfService);
                        Log.i(TAG, "******* subscribe ******** " + topics[0]);
                        //Toast.makeText(getApplication(),"注册成功",Toast.LENGTH_SHORT).show();
                        subscribed = true;
                        if (subscribed == false) {
                            broadcastServiceStatus("Unable to subscribe");           //notifyUser("Unable to subscribe", "MQTT", "Unable to subscribe");
                        }
                    } catch (MqttNotConnectedException e) {
                        Log.e("mqtt", "subscribe failed - MQTT not connected", e);
                    } catch (IllegalArgumentException e) {
                        Log.e("mqtt", "subscribe failed - illegal argument", e);
                    } catch (MqttException e) {
                        Log.e("mqtt", "subscribe failed - MQTT exception", e);
                    }
                }
            }.start();
        }
    }

    private void disconnectFromBroker() {
        // if we've been waiting for an Internet connection, this can be
        //  cancelled - we don't need to be told when we're connected now
        try {
            if (netConnReceiver != null) {
                unregisterReceiver(netConnReceiver);
                netConnReceiver = null;
            }

            if (pingSender != null) {
                unregisterReceiver(pingSender);
                pingSender = null;
            }
        } catch (Exception eee) {
            // probably because we hadn't registered it
            Log.e("mqtt", "unregister failed", eee);
        }


        if (mqttClient != null) {
            new Thread() {
                @Override
                public void run() {
                    try {
                        mqttClient.disconnect();
                    } catch (MqttPersistenceException e) {
                        Log.e("mqtt", "disconnect failed - persistence exception", e);
                    } finally {
                        mqttClient = null;
                    }
                }
            }.start();
        }
        NotificationManager nm = (NotificationManager) getSystemService(NOTIFICATION_SERVICE);
        nm.cancelAll();
    }

    private boolean isAlreadyConnected() {
        return ((mqttClient != null) && (mqttClient.isConnected() == true));
    }


    private class BackgroundDataChangeIntentReceiver extends BroadcastReceiver {
        @Override
        public void onReceive(Context ctx, Intent intent) {
            // we protect against the phone switching off while we're doing this
            //  by requesting a wake lock - we request the minimum possible wake
            //  lock - just enough to keep the CPU running until we've finished
            Log.i(TAG, "20170427 BackgroundDataChangeIntentReceiver");
            PowerManager pm = (PowerManager) getSystemService(POWER_SERVICE);
            @SuppressLint("InvalidWakeLockTag") WakeLock wl = pm.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "MQTT");
            wl.acquire();

            ConnectivityManager cm = (ConnectivityManager) getSystemService(CONNECTIVITY_SERVICE);
            if (cm.getBackgroundDataSetting()) {
                // user has allowed background data - we start again - picking
                //  up where we left off in handleStart before
                defineConnectionToBroker(brokerHostName);
                Log.i(TAG, "BackgroundDataChangeIntentReceiver");
                handleStart(intent, 0);
            } else {
                // user has disabled background data
                connectionStatus = MQTTConnectionStatus.NOTCONNECTED_DATADISABLED;

                // update the app to show that the connection has been disabled
                broadcastServiceStatus("Not connected - background data disabled");

                // disconnect from the broker
                disconnectFromBroker();
            }

            // we're finished - if the phone is switched off, it's okay for the CPU
            //  to sleep now
            wl.release();
        }
    }

    /*
     * Called in response to a change in network connection - after losing a
     *  connection to the server, this allows us to wait until we have a usable
     *  data connection again
     */
    private class NetworkConnectionIntentReceiver extends BroadcastReceiver {
        @Override
        public void onReceive(Context ctx, Intent intent) {
            // we protect against the phone switching off while we're doing this
            //  by requesting a wake lock - we request the minimum possible wake
            //  lock - just enough to keep the CPU running until we've finished
            Log.i(TAG, " 20170427 NetworkConnectionIntentReceiver");
            PowerManager pm = (PowerManager) getSystemService(POWER_SERVICE);
            @SuppressLint("InvalidWakeLockTag") WakeLock wl = pm.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "MQTT");
            wl.acquire();

            if (isOnline()) {
                // we have an internet connection - have another try at connecting
                if (connectToBroker()) {
                    // we subscribe to a topic - registering to onReceiveFile push
                    //  notifications with a particular key
                    subscribeToTopic(subTopic);
                }
            }

            // we're finished - if the phone is switched off, it's okay for the CPU
            //  to sleep now
            wl.release();
        }
    }

    /*
     * Schedule the next time that you want the phone to wake up and ping the
     *  message broker server
     */
    private void scheduleNextPing() {
        Log.i(TAG, "scheduleNextPing");
        PendingIntent pendingIntent = PendingIntent.getBroadcast(this, 0,
                new Intent(MQTT_PING_ACTION),
                PendingIntent.FLAG_UPDATE_CURRENT);
        Calendar wakeUpTime = Calendar.getInstance();
        wakeUpTime.add(Calendar.SECOND, keepAliveSeconds);

        AlarmManager aMgr = (AlarmManager) getSystemService(ALARM_SERVICE);
        aMgr.set(AlarmManager.RTC_WAKEUP,
                wakeUpTime.getTimeInMillis(),
                pendingIntent);
    }

    /*
     * Used to implement a keep-alive protocol at this Service level - it sends
     *  a PING message to the server, then schedules another ping after an
     *  interval defined by keepAliveSeconds
     */
    public class PingSender extends BroadcastReceiver {
        @Override
        public void onReceive(Context context, Intent intent) {

            try {
                mqttClient.ping();
            } catch (MqttException e) {
                // if something goes wrong, it should result in connectionLost
                //  being called, so we will handle it there
                Log.e("mqtt", "ping failed - MQTT exception", e);

                // assume the client connection is broken - trash it
                try {
                    mqttClient.disconnect();
                } catch (MqttPersistenceException e1) {
                    Log.e("mqtt", "disconnect failed - persistence exception", e1);
                }

                // reconnect
                if (connectToBroker()) {
                    subscribeToTopic(subTopic);
                }
            }

            // start the next keep alive period
            scheduleNextPing();
        }
    }

    /************************************************************************/
    /*   APP SPECIFIC - stuff that would vary for different uses of MQTT	*/
    /************************************************************************/

    //  apps that handle very small amounts of data - e.g. updates and
    //   notifications that don't need to be persisted if the app / phone
    //   is restarted etc. may find it acceptable to store this data in a
    //   variable in the Service
    //  that's what I'm doing in this sample: storing it in a local hashtable
    //  if you are handling larger amounts of data, and/or need the data to
    //   be persisted even if the app and/or phone is restarted, then
    //   you need to store the data somewhere safely
    //  see http://developer.android.com/guide/topics/data/data-storage.html
    //   for your storage options - the best choice depends on your needs

    // stored internally

    private Hashtable<String, String> dataCache = new Hashtable<String, String>();

    private boolean addReceivedMessageToStore(String key, String value) {
        String previousValue = null;

        if (value.length() == 0) {
            previousValue = dataCache.remove(key);
        } else {
            previousValue = dataCache.put(key, value);
        }

        // is this a new value? or am I receiving something I already knew?
        //  we return true if this is something new
        return ((previousValue == null) ||
                (previousValue.equals(value) == false));
    }

    // provide a public interface, so Activities that bind to the Service can
    //  request access to previously received messages

    public void rebroadcastReceivedMessages() {
        Enumeration<String> e = dataCache.keys();
        while (e.hasMoreElements()) {
            String nextKey = e.nextElement();
            String nextValue = dataCache.get(nextKey);
            Log.i(TAG, "rebroadcastReceivedMessages");
            //broadcastReceivedMessage(nextKey, nextValue);
            msgArrived(nextKey, nextValue);
        }
    }

    /************************************************************************/
    /*	METHODS - internal utility methods								*/

    /************************************************************************/

    private String generateClientId() {
        // generate a unique client id if we haven't done so before, otherwise
        //   re-use the one we already have

        if (mqttClientId == null) {
            // generate a unique client ID - I'm basing this on a combination of
            //  the phone device id and the current timestamp
            String timestamp = "" + (new Date()).getTime();
            String android_id = Settings.System.getString(getContentResolver(),
                    Secure.ANDROID_ID);
            mqttClientId = timestamp + android_id;

            // truncate - MQTT spec doesn't allow client ids longer than 23 chars
            if (mqttClientId.length() > MAX_MQTT_CLIENTID_LENGTH) {
                mqttClientId = mqttClientId.substring(0, MAX_MQTT_CLIENTID_LENGTH);
            }
        }
        Log.i(TAG, "mqttClientId: " + mqttClientId);
        return mqttClientId;
    }

    private boolean isOnline() {
        ConnectivityManager cm = (ConnectivityManager) getSystemService(CONNECTIVITY_SERVICE);
        if (cm.getActiveNetworkInfo() != null &&
                cm.getActiveNetworkInfo().isAvailable() &&
                cm.getActiveNetworkInfo().isConnected()) {
            return true;
        }
        return false;
    }

    private void publishMqttMessage(final String topic, final String msg) {

        if (mqttClient.isConnected()) {

            new Thread() {
                @Override
                public void run() {
                    super.run();
                    try {
                        mqttClient.publish(topic, msg.getBytes(), 2, false);
                        // Log.i(TAG, subTopic + " PUBLISH MESSAGE TO: " + topic + "\r\n" + msg);

                    } catch (MqttException e) {
                        Log.e(TAG, "publishMqttMessage error: " + e.getMessage());

                        //e.printStackTrace();
                    }


                }
            }.start();
        } else {
            Log.i(TAG, "disconnected! publish fail!!!");
        }
    }

    /*
     *   callback - called when we onReceiveFile a message from the server
     */
    public void publishArrived(String topic, byte[] payloadbytes, int qos, boolean retained) {

        String messageBody = new String(payloadbytes);

        if (addReceivedMessageToStore(topic, messageBody)) {

            msgArrived(topic, messageBody);
        }
        scheduleNextPing();
    }


    protected void msgArrived(String topic, String msg) {
        Log.i(TAG, subTopic + " MESSAGE ARRIVED: " + topic + " : " + msg);
        JSONObject json = new JSONObject();
        json.put("topic", topic);
        json.put("payload", msg);
        Intent intent = new Intent();
        intent.setAction(Constants.ACTION_MQTT_SUB_MSG);
        intent.putExtra("msg", json.toJSONString());

        sendBroadcast(intent);

//        MqttMsg mqttMsg = JsonUtils.toBean(msg, MqttMsg.class);
//        String action = mqttMsg.getAction();
//        if (action.isEmpty()) {
//            return;
//        }
//        String clz = Utils.getPkgName(getApplicationContext()) + ".action." + action;
//        Log.i(TAG, clz);
//        Class<?> cl = null;
//        try {
//            cl = Class.forName(clz);
//            Constructor constructor = cl.getConstructor(Context.class, String.class);
//            Object obj = constructor.newInstance(getApplicationContext(), msg);
//        } catch (Exception e) {
//
//        }
    }

    private class MsgReceiver extends BroadcastReceiver {
        @Override
        public void onReceive(Context context, Intent intent) {

            try {
                int flag = intent.getIntExtra("MessageType", -1);
                Log.i(TAG, "消息到达"+flag);
                switch (flag) {
                    case 0:

                        initMqttService();

                        new Thread(new Runnable() {
                            @Override
                            public void run() {
                                handleStart();
                            }
                        }, "MQTTservice").start();
                        break;
                    case 1:
                        final String msg = intent.getStringExtra("MqttMessage");
                        final String topic = intent.getStringExtra("Topic");
                        publishMqttMessage(topic, msg);
                        break;
                    default:
                        break;
                }

            } catch (Exception e) {
                Log.e(TAG, "MsgReceiver error: " + e.getMessage());
            }
        }
    }

}

